package com.lazyboyl.big.data.read.util;




import com.lazyboyl.big.data.read.entity.StartEndPair;
import com.lazyboyl.big.data.read.handle.IHandle;

import java.io.*;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author linzf
 * @since 2021-07-29
 * 类描述： 超大文件读取的实现
 */
public class BigFileReader {

    /**
     * 当前的数据需要划分成多少块来进行读取，例如10G的数据要划分成4个2.5G那就将dataBlockSize设置为4即可
     */
    private int dataBlockSize;
    /**
     * 编码方式
     */
    private String charset;
    /**
     * 每次从流中读取的数据的大小
     */
    private int bufferSize;
    /**
     * 一行数据读取完成以后需要进行的一些操作
     */
    private IHandle handle;
    /**
     * 根据dataBlockSize和fileLength来决定每个分片的数据的大小
     */
    private long blockLength;
    /**
     * 待读取文件的总大小
     */
    private long fileLength;
    /**
     * RandomAccessFile可以自由访问文件的任意位置,因此此处使用RandomAccessFile来读取数据
     */
    private RandomAccessFile rAccessFile;
    /**
     * 用于存放分组的数据
     */
    private List<StartEndPair> startEndPairList = new ArrayList<>();

    /**
     * 并发的方式读取数据
     *
     */
    public void concurrentStart() {
        /**
         * 初始化线程池
         */
        ExecutorService es = new ThreadPoolExecutor(5, 50, 0, TimeUnit.SECONDS, new SynchronousQueue<>(),
                r -> {
                    Thread t = new Thread(r);
                    t.setDaemon(true);
                    t.setName("BigFileReader" + r);
                    return t;
                }, (runnable, executor) -> {
        });
        for (StartEndPair startEndPair : startEndPairList) {
            es.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        readTxt(startEndPair);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

    /**
     * 功能描述： 单线程的方式启动数据的读取
     */
    public void singleStart() {
        for (StartEndPair startEndPair : startEndPairList) {
            try {
                readTxt(startEndPair);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 功能描述：初始化大文件读取对象
     *
     * @param file          文件
     * @param handle        自定义执行handle
     * @param charset       文件编码
     * @param bufferSize    每次读取数据的大小
     * @param dataBlockSize 将数据分为多少块
     */
    public BigFileReader(File file, IHandle handle, String charset, int bufferSize, int dataBlockSize) {
        this.fileLength = file.length();
        this.blockLength = this.fileLength / dataBlockSize;
        this.handle = handle;
        this.charset = charset;
        this.bufferSize = bufferSize;
        this.dataBlockSize = dataBlockSize;
        try {
            this.rAccessFile = new RandomAccessFile(file, "r");
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }
        this.init();
    }

    /**
     * 功能描述： 实现数据的读取
     *
     * @param startEndPair 数据读取的开始和结束位置的实体对象
     * @throws IOException
     */
    public void readTxt(StartEndPair startEndPair) throws IOException {
        System.out.println("startTime=>" + System.currentTimeMillis());
        MappedByteBuffer mapBuffer = this.rAccessFile.getChannel().map(FileChannel.MapMode.READ_ONLY, startEndPair.getStart(), startEndPair.getEnd() - startEndPair.getStart());
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        byte[] readBuff = new byte[bufferSize];
        for (int offset = 0; offset < startEndPair.getEnd() - startEndPair.getStart(); offset += bufferSize) {
            int readLength;
            if (offset + bufferSize <= (startEndPair.getEnd() - startEndPair.getStart())) {
                readLength = bufferSize;
            } else {
                readLength = (int) (startEndPair.getEnd() - startEndPair.getStart() - offset);
            }
            mapBuffer.get(readBuff, 0, readLength);
            for (int i = 0; i < readLength; i++) {
                byte tmp1 = readBuff[i];
                if (tmp1 == '\n' || tmp1 == '\r') {
                    handle.readLine(new String(bos.toByteArray(), charset));
                    bos.reset();
                } else {
                    bos.write(tmp1);
                }
            }
        }
        bos.close();
        mapBuffer.clear();
        this.rAccessFile.seek(startEndPair.getEnd() + 1);
        System.out.println("endTime=>" + System.currentTimeMillis());
    }

    /**
     * 功能描述： 实现数据的动态划分，因为根据我们的初始划分，可能存在刚好数据的末尾的位置并不是我们这一行的结束
     * 因此我们需要将数据的末尾位置定位到刚刚好换行的位置。
     */
    private void init() {
        try {
            long startSize = 0;
            long endPosition = this.fileLength / this.dataBlockSize;
            recursionPartitioning(startSize, endPosition, this.dataBlockSize);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 功能描述： 获取分片的位置
     *
     * @param startPoint 开始位置
     * @param endPoint   结束位置
     * @return 返回相应的数据的读取的起始和结束位置
     * @throws IOException
     */
    private void recursionPartitioning(long startPoint, long endPoint, long recursionNum) throws IOException {
        recursionNum = recursionNum - 1;
        StartEndPair startEndPair = new StartEndPair();
        startEndPair.setStart(startPoint);
        if (recursionNum == 0) {
            startEndPair.setEnd(this.fileLength);
            startEndPairList.add(startEndPair);
            return;
        }
        this.rAccessFile.seek(endPoint);
        byte tmp = (byte) rAccessFile.read();
        while (tmp != '\n' && tmp != '\r') {
            endPoint++;
            if (endPoint >= fileLength - 1) {
                endPoint = fileLength - 1;
                break;
            }
            rAccessFile.seek(endPoint);
            tmp = (byte) rAccessFile.read();
        }
        startEndPair.setEnd(endPoint + 1);
        startEndPairList.add(startEndPair);
        if (recursionNum > 0) {
            recursionPartitioning(endPoint + 1, endPoint + 1 + this.blockLength, recursionNum);
        }
    }

}
